package com.cheng.im.service.massage.service.impl;

import com.cheng.im.common.bean.msg.ProtoMsgV1;
import com.cheng.im.service.conversation.service.ImConversationSetService;
import com.cheng.im.service.group.service.ImGroupMemberService;
import com.cheng.im.service.massage.mapper.ImMessageBodyMapper;
import com.cheng.im.service.mq.RocketMQProduce;
import com.cheng.im.service.redis.seq.RedisSeq;
import com.cheng.im.service.util.SnowflakeIdWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Slf4j
@Service
public class MessageSyncService {

    @Resource
    RocketMQProduce rocketMQProduce;

    @Resource
    ImConversationSetService conversationService;

    @Resource
    RedisTemplate redisTemplate;

    @Resource
    ImMessageBodyMapper imMessageBodyMapper;

    @Resource
    RedisSeq redisSeq;

    @Resource
    SnowflakeIdWorker snowflakeIdWorker;

    @Resource
    ImGroupMemberService imGroupMemberService;


    public void receiveMark(ProtoMsgV1.Message msgReceiveAck){

        rocketMQProduce.dispatchMessage(msgReceiveAck.getMessageReceiveAck().getTo(), msgReceiveAck);
    }

    /**
     * @description: 消息已读。更新会话的seq，通知在线的同步端发送指定command ，发送已读回执通知对方（消息发起方）我已读
     * @param
     * @return void
     * @author lld
     */
    public void readMark(ProtoMsgV1.Message msgRead) {
        conversationService.messageMarkRead(msgRead);

        syncToSender(msgRead);
        //发送给对方
        rocketMQProduce.sendToUser(msgRead.getMessageRead().getTo(), msgRead, true);
    }

    private void syncToSender(ProtoMsgV1.Message msgRead){
        //发送给自己的其他端
        rocketMQProduce.sendToUserExceptClient(msgRead.getMessageRead().getFrom(), msgRead);
    }

    public void groupReadMark(ProtoMsgV1.Message msgRead) {
        conversationService.messageMarkRead(msgRead);

        syncToSender(msgRead);

        if(!msgRead.getMessageRead().getFrom().equals(msgRead.getMessageRead().getTo())){
            rocketMQProduce.sendToUser(msgRead.getMessageRead().getTo(), msgRead, true);
        }
    }

//    public ResponseVO syncOfflineMessage(SyncReq req) {
//
//        SyncResp<OfflineMessageContent> resp = new SyncResp<>();
//
//        String key = req.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + req.getOperater();
//        //获取最大的seq
//        Long maxSeq = 0L;
//        ZSetOperations zSetOperations = redisTemplate.opsForZSet();
//        Set set = zSetOperations.reverseRangeWithScores(key, 0, 0);
//        if(!CollectionUtils.isEmpty(set)){
//            List list = new ArrayList(set);
//            DefaultTypedTuple o = (DefaultTypedTuple) list.get(0);
//            maxSeq = o.getScore().longValue();
//        }
//
//        List<OfflineMessageContent> respList = new ArrayList<>();
//        resp.setMaxSequence(maxSeq);
//
//        Set<ZSetOperations.TypedTuple> querySet = zSetOperations.rangeByScoreWithScores(key,
//                req.getLastSequence(), maxSeq, 0, req.getMaxLimit());
//        for (ZSetOperations.TypedTuple<String> typedTuple : querySet) {
//            String value = typedTuple.getValue();
//            OfflineMessageContent offlineMessageContent = JSONObject.parseObject(value, OfflineMessageContent.class);
//            respList.add(offlineMessageContent);
//        }
//        resp.setDataList(respList);
//
//        if(!CollectionUtils.isEmpty(respList)){
//            OfflineMessageContent offlineMessageContent = respList.get(respList.size() - 1);
//            resp.setCompleted(maxSeq <= offlineMessageContent.getMessageKey());
//        }
//
//        return ResponseVO.successResponse(resp);
//    }

    //修改历史消息的状态
    //修改离线消息的状态
    //ack给发送方
    //发送给同步端
    //分发给消息的接收方
//    public void recallMessage(RecallMessageContent content) {
//
//        Long messageTime = content.getMessageTime();
//        Long now = System.currentTimeMillis();
//
//        RecallMessageNotifyPack pack = new RecallMessageNotifyPack();
//        BeanUtils.copyProperties(content,pack);
//
//        if(120000L < now - messageTime){
//            recallAck(pack,ResponseVO.errorResponse(MessageErrorCode.MESSAGE_RECALL_TIME_OUT),content);
//            return;
//        }
//
//        QueryWrapper<ImMessageBodyEntity> query = new QueryWrapper<>();
//        query.eq("app_id",content.getAppId());
//        query.eq("message_key",content.getMessageKey());
//        ImMessageBodyEntity body = imMessageBodyMapper.selectOne(query);
//
//        if(body == null){
//            //TODO ack失败 不存在的消息不能撤回
//            recallAck(pack,ResponseVO.errorResponse(MessageErrorCode.MESSAGEBODY_IS_NOT_EXIST),content);
//            return;
//        }
//
//        if(body.getDelFlag() == DelFlagEnum.DELETE.getCode()){
//            recallAck(pack,ResponseVO.errorResponse(MessageErrorCode.MESSAGE_IS_RECALLED),content);
//
//            return;
//        }
//
//        body.setDelFlag(DelFlagEnum.DELETE.getCode());
//        imMessageBodyMapper.update(body,query);
//
//        if(content.getConversationType() == ConversationTypeEnum.P2P.getCode()){
//
//            // 找到fromId的队列
//            String fromKey = content.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + content.getFromId();
//            // 找到toId的队列
//            String toKey = content.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + content.getToId();
//
//            OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
//            BeanUtils.copyProperties(content,offlineMessageContent);
//            offlineMessageContent.setDelFlag(DelFlagEnum.DELETE.getCode());
//            offlineMessageContent.setMessageKey(content.getMessageKey());
//            offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
//            offlineMessageContent.setConversationId(conversationService.convertConversationId(offlineMessageContent.getConversationType()
//                    ,content.getFromId(),content.getToId()));
//            offlineMessageContent.setMessageBody(body.getMessageBody());
//
//            long seq = redisSeq.doGetSeq(content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(content.getFromId(),content.getToId()));
//            offlineMessageContent.setMessageSequence(seq);
//
//            long messageKey = SnowflakeIdWorker.nextId();
//
//            redisTemplate.opsForZSet().add(fromKey,JSONObject.toJSONString(offlineMessageContent),messageKey);
//            redisTemplate.opsForZSet().add(toKey,JSONObject.toJSONString(offlineMessageContent),messageKey);
//
//            //ack
//            recallAck(pack,ResponseVO.successResponse(),content);
//            //分发给同步端
//            messageProducer.sendToUserExceptClient(content.getFromId(),
//                    MessageCommand.MSG_RECALL_NOTIFY,pack,content);
//            //分发给对方
//            messageProducer.sendToUser(content.getToId(),MessageCommand.MSG_RECALL_NOTIFY,
//                    pack,content.getAppId());
//        }else{
//            List<String> groupMemberId = imGroupMemberService.getGroupMemberId(content.getToId(), content.getAppId());
//            long seq = redisSeq.doGetSeq(content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(content.getFromId(),content.getToId()));
//            //ack
//            recallAck(pack,ResponseVO.successResponse(),content);
//            //发送给同步端
//            messageProducer.sendToUserExceptClient(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack
//                    , content);
//            for (String memberId : groupMemberId) {
//                String toKey = content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + memberId;
//                OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
//                offlineMessageContent.setDelFlag(DelFlagEnum.DELETE.getCode());
//                BeanUtils.copyProperties(content,offlineMessageContent);
//                offlineMessageContent.setConversationType(ConversationTypeEnum.GROUP.getCode());
//                offlineMessageContent.setConversationId(conversationService.convertConversationId(offlineMessageContent.getConversationType()
//                        ,content.getFromId(),content.getToId()));
//                offlineMessageContent.setMessageBody(body.getMessageBody());
//                offlineMessageContent.setMessageSequence(seq);
//                redisTemplate.opsForZSet().add(toKey,JSONObject.toJSONString(offlineMessageContent),seq);
//
//                groupMessageProducer.producer(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack,content);
//            }
//        }
//
//    }

//    private void recallAck(RecallMessageNotifyPack recallPack, ResponseVO<Object> success, ClientInfo clientInfo) {
//        ResponseVO<Object> wrappedResp = success;
//        messageProducer.sendToUser(recallPack.getFromId(),
//                MessageCommand.MSG_RECALL_ACK, wrappedResp, clientInfo);
//    }
}
